package com.jackrain.clouddata.sys.Listenter;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.impl.util.MsgConvertUtil;
import com.jackrain.nea.config.PropertiesConf;
import com.jackrain.nea.util.ApplicationContextHandle;
import com.jackrain.nea.util.InstanceUtil;
import com.jackrain.nea.util.Tools;
import com.jackrain.nea.util.tlock.RedisReentrantLock;
import com.jackrain.nea.web.query.QueryEngine;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.PrintWriter;
import java.io.Serializable;
import java.io.StringWriter;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 消费云数据消息
 */

@Slf4j
public class MessageListenerAdapter implements MessageListener {
    private static final Logger logger = LoggerFactory.getLogger(MessageListenerAdapter.class);

    public MessageListenerAdapter() {

    }
    PropertiesConf pconf = ApplicationContextHandle.getBean(PropertiesConf.class);

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        String[] tag = message.getTag().split(":");
        String client = null;
        String topic = null;
        if(tag.length > 0){
            client = tag[0];
        }
        //TODO 日志相关变量
        //String selectSql = "select count(1) from ons_sendmonitor where msgkey=? and state=1";

        //TODO 多条解决方案 insert ons_sendmonitor by every time consumre
        String updateSql = "update ons_sendmonitor set consumer_error=?,pdesc=?,state=2 where msgkey=?";

        String uniqKey = message.getKey();

        log.debug("process uniqKey->"+uniqKey);
        //日志流
        StringWriter sw = new StringWriter();
        PrintWriter pw = new PrintWriter(sw);

        //TODO Redis上锁验证
        RedisReentrantLock lock = new RedisReentrantLock("MQ_MESSAGE:"+uniqKey);
        try {

            if (lock.tryLock(500L, TimeUnit.MILLISECONDS)) {
                Serializable ser = MsgConvertUtil.objectDeserialize(message.getBody());
                JSONObject paramObject = JSONObject.parseObject((String) ser);
                String classname = paramObject.getString("classname");
                Object o = paramObject.get("data");

//                log.debug("process classname  start->" + classname);
//                log.debug("process o ->" + o.toString());

                MessageProcess messageProcess = (MessageProcess) ApplicationContextHandle.getBean(classname);
                messageProcess.process(o, client);
                QueryEngine.getInstance().
                        executeUpdate(updateSql,
                                new Object[]{"N","consume successfully",uniqKey});

                log.debug("process classname  end->" + classname);
                return Action.CommitMessage;
            }else {
                // 出现异常,再次消费
                log.debug("process lockfail return later!!!!!!!");
                return Action.ReconsumeLater;
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            e.printStackTrace(pw);
            QueryEngine.getInstance().
                    executeUpdate(updateSql,
                            new Object[]{"Y", sw.toString(),uniqKey});

            return Action.CommitMessage;
        }finally {
            lock.unlock();
        }
    }
}